package org.ricks.gateway.cluster;

import io.scalecube.cluster.Cluster;
import io.scalecube.cluster.ClusterConfig;
import io.scalecube.cluster.ClusterImpl;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.ricks.gateway.utils.NetworkKit;
import org.ricks.ioc.utils.Logger;

import java.util.List;


@Setter
@Getter
@Accessors(chain = true)
@Slf4j
public final class BrokerClusterManager {
    /** broker （游戏网关）唯一 id */
    private String brokerId;
    /** broker 端口（游戏网关端口） */
    private int port;
    /** Gossip listen port 监听端口 */
    private int gossipListenPort;

    /**
     * 种子节点地址
     * <pre>
     *     格式： ip:port
     * </pre>
     */
    private List<String> seedAddress;

    private Cluster cluster;

    private String name;

    private EnhancedGossipProtocol messageHandler;

    public BrokerClusterManager() {
    }

    public void start() {
        final String localIp = NetworkKit.LOCAL_IP;
        Broker localBroker = getLocalBroker(localIp);

        this.name = String.format("ioGameCluster-%d-%d-%s", port, gossipListenPort, localIp);

        // 创建元数据解析器
        MetadataResolver resolver = () ->
                cluster.metadata()
                        .filter(BrokerClusterMetadata.class::isInstance)
                        .map(BrokerClusterMetadata.class::cast)
                        .map(BrokerClusterMetadata::getLocalBroker);
        this.messageHandler = new EnhancedGossipProtocol(resolver);

        ClusterConfig config = new ClusterConfig()
                .memberAlias(name)
                .metadata(new BrokerClusterMetadata(name, localBroker))
                .externalHost(NetworkKit.LOCAL_IP)
                .externalPort(gossipListenPort)
                .membership(cfg -> cfg
                        .seedMembers(seedAddress)
                        .syncInterval(5_000))
                .transport(cfg -> cfg
                        .port(gossipListenPort));
//                        .messageCodec(new JProtobufMessageCodec(CodecRegistry.createDefault())));

        this.cluster = new ClusterImpl()
                .config(_ -> config)
                .handler(messageHandler)
                .transportFactory(SmartSocketTransportFactory::new)
                .startAwait();
        cluster.members().forEach(member -> Logger.info("member:  {} - {} - {}", member.id(), member.address(),
                member.alias()));
        Logger.info("集群搭建成功  {}" , cluster.members().size());

//        Map<String, Broker> brokers = this.messageHandler.brokers;
//        brokers.put(localBroker.getClusterAddress(), localBroker);
    }

//    public BrokerClusterMessage getBrokerClusterMessage() {
//        return this.messageHandler.getBrokerClusterMessage();
//    }

    private Broker getLocalBroker(String localIp) {
        String clusterAddress = localIp + ":" + this.gossipListenPort;
        String brokerAddress = localIp + ":" + this.port;
        Broker broker =  new Broker(localIp);
        broker .setId(this.brokerId);
        broker  .setPort(this.port);
        broker  .setBrokerAddress(brokerAddress);
        broker  .setClusterAddress(clusterAddress);
        return broker;
    }
}
